package com.edata.bigdata.basic;

import org.apache.spark.sql.types.*;

import java.io.Serializable;
import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.lang.reflect.Type;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

public final class Commons implements Serializable {

    public static DataType toSparkTypeMapper(Type type) {
        return switch (type.getTypeName()) {
            /*
             * 基础类型
             * */
            case "int", "java.lang.Integer" -> DataTypes.IntegerType;
            case "long", "java.lang.Long" -> DataTypes.LongType;
            case "double", "java.lang.Double" -> DataTypes.DoubleType;
            case "boolean", "java.lang.Boolean" -> DataTypes.BooleanType;
            case "short", "java.lang.Short" -> DataTypes.ShortType;
            case "byte", "java.lang.Byte" -> DataTypes.ByteType;
            case "float", "java.lang.Float" -> DataTypes.FloatType;
            case "java.lang.String", "java.sql.Time" -> DataTypes.StringType;
            case "java.math.BigDecimal" -> DataTypes.createDecimalType(38, 18);
            /*
             * 日期时间类型
             * */
            case "java.util.Date" -> DataTypes.TimestampType;
            case "java.sql.Date" -> DataTypes.DateType;
            case "java.sql.Timestamp" -> DataTypes.TimestampType;
            /*
             * 数组类型
             * */
            case "[B" -> DataTypes.BinaryType;
            default -> DataTypes.StringType;
        };
    }

    public static <T> StructType createDataFrameSchema(Class<T> clazz) {
        List<StructField> sfs = new ArrayList<>();
        for (Field field : clazz.getDeclaredFields()) {
            String fieldName = field.getName();
            DataType fieldType = Commons.toSparkTypeMapper(field.getGenericType());
            sfs.add(DataTypes.createStructField(fieldName, fieldType, true, Metadata.empty()));
        }
        return DataTypes.createStructType(sfs);
    }

    public static <T> String createUnPreparedUpsertSQLByClazz(String conflictField, Class<T> clazz) {
        List<String> fieldNames = new ArrayList<>();
        List<String> insertValuesPlaceHolders = new ArrayList<>();
        List<String> updateValuesPlaceHolders = new ArrayList<>();
        String tableName = clazz.getSimpleName().toLowerCase();
        for (Field field : clazz.getDeclaredFields()) {
            fieldNames.add(field.getName());
            insertValuesPlaceHolders.add("?");
            if (!conflictField.equals(field.getName())) {
                updateValuesPlaceHolders.add(field.getName() + "=" + "EXCLUDED." + field.getName());
            }
        }
        return "INSERT INTO " + tableName +
                "(" + String.join(",", fieldNames) + ")" +
                " VALUES (" + String.join(",", insertValuesPlaceHolders) + ")" +
                " ON CONFLICT (" + conflictField + ") DO UPDATE SET " +
                String.join(",", updateValuesPlaceHolders);
    }

    public static <T> void setPreparedStatementParameters(PreparedStatement stmt, T record, Class<T> clazz) throws SQLException, IllegalAccessException {
        Field[] fields = clazz.getDeclaredFields();
        int paramIndex = 1;
        for (Field field : fields) {
            field.setAccessible(true);
            Object value = field.get(record);
            if (value == null) {
                stmt.setObject(paramIndex++, null);
            } else if (value instanceof String) {
                stmt.setString(paramIndex++, (String) value);
            } else if (value instanceof Integer) {
                stmt.setInt(paramIndex++, (Integer) value);
            } else if (value instanceof Long) {
                stmt.setLong(paramIndex++, (Long) value);
            } else if (value instanceof java.util.Date) {
                stmt.setTimestamp(paramIndex++, new java.sql.Timestamp(((java.util.Date) value).getTime()));
            } else if (value instanceof java.time.LocalDateTime) {
                stmt.setTimestamp(paramIndex++, java.sql.Timestamp.valueOf((java.time.LocalDateTime) value));
            } else {
                stmt.setObject(paramIndex++, value);
            }
        }
    }

    public static Field findFieldFromBeanByAnnotation(Class<?> clazz, Class<? extends Annotation> annotationType) throws NoSuchFieldException {
        for (Field field : clazz.getDeclaredFields()) {
            if (field.isAnnotationPresent(annotationType)) {
                return field;
            }
        }
        throw new NoSuchFieldException("找不到目标字段" + annotationType.getSimpleName());
    }


}
